package cn.itcast.job;

import cn.itcast.avro.AvroDeserializerSchema;
import cn.itcast.avro.SseAvro;
import cn.itcast.avro.SzseAvro;
import cn.itcast.bean.CleanBean;
import cn.itcast.config.QuotConfig;
import cn.itcast.function.IndexQuotMinHdfsTask;
import cn.itcast.function.IndexQuotMinTask;
import cn.itcast.map.SseMap;
import cn.itcast.map.SzseMap;
import cn.itcast.task.IndexQuotSecTask;
import cn.itcast.util.QuotUtil;
import org.apache.flink.api.common.functions.FilterFunction;
import org.apache.flink.streaming.api.TimeCharacteristic;
import org.apache.flink.streaming.api.datastream.DataStream;
import org.apache.flink.streaming.api.environment.StreamExecutionEnvironment;
import org.apache.flink.streaming.api.functions.timestamps.BoundedOutOfOrdernessTimestampExtractor;
import org.apache.flink.streaming.api.windowing.time.Time;
import org.apache.flink.streaming.connectors.kafka.FlinkKafkaConsumer011;

import java.util.Properties;

public class IndexStream {
    public static void main(String[] args) {
        //2.获取流处理执行环境
        StreamExecutionEnvironment env = StreamExecutionEnvironment.getExecutionEnvironment();
        //3.设置事件时间、并行度
        env.setParallelism(1);
        env.setStreamTimeCharacteristic(TimeCharacteristic.EventTime);
        //设置检查点机制
      /*  env.enableCheckpointing(5000);
        env.setStateBackend(new FsStateBackend("hdfs://node01:8020/checkpoint/stock"));
        env.getCheckpointConfig().setCheckpointingMode(CheckpointingMode.EXACTLY_ONCE);
        env.getCheckpointConfig().setCheckpointTimeout(60000);
        env.getCheckpointConfig().setFailOnCheckpointingErrors(false);
        env.getCheckpointConfig().setMaxConcurrentCheckpoints(1);
        env.getCheckpointConfig().enableExternalizedCheckpoints(CheckpointConfig.ExternalizedCheckpointCleanup.DELETE_ON_CANCELLATION);
        //5.设置重启机制
        env.setRestartStrategy(RestartStrategies.fixedDelayRestart(3, Time.seconds(5)));*/
/*        Properties properties = new Properties();
        properties.setProperty("bootstrap.servers", QuotConfig.config.getProperty("bootstrap.servers"));
        properties.setProperty("group.id", QuotConfig.config.getProperty("group.id"));*/
        Properties props=new Properties();
        props.setProperty("bootstrap.servers", QuotConfig.config.getProperty("bootstrap.servers"));
        props.setProperty("group.id", QuotConfig.config.getProperty("group.id"));
        FlinkKafkaConsumer011<SseAvro> sseKafkaData = new FlinkKafkaConsumer011<SseAvro>(QuotConfig.config.getProperty("sse.topic"),new AvroDeserializerSchema<>(QuotConfig.config.getProperty("sse.topic")),props);
        FlinkKafkaConsumer011<SzseAvro>  szseKafkaData = new FlinkKafkaConsumer011<SzseAvro>(QuotConfig.config.getProperty("szse.topic"),new AvroDeserializerSchema<>(QuotConfig.config.getProperty("szse.topic")),props);
        sseKafkaData.setStartFromEarliest();
        szseKafkaData.setStartFromEarliest();
        DataStream<SseAvro> sseSourceData = env.addSource(sseKafkaData);
        DataStream<SzseAvro> szseSourceData = env.addSource(szseKafkaData);
      //  sseSourceData.print();
      //  szseSourceData.print();
       // 7.数据过滤（时间和null字段）
        DataStream<SseAvro>  sseFileData = sseSourceData.filter(new FilterFunction<SseAvro>() {
            @Override
            public boolean filter(SseAvro value) throws Exception {
                return QuotUtil.checkTime(value) && QuotUtil.checkData(value);
            }
        });
        DataStream<SzseAvro> szseFileData = szseSourceData.filter(new FilterFunction<SzseAvro>() {
            @Override
            public boolean filter(SzseAvro value) throws Exception {
                return QuotUtil.checkTime(value) && QuotUtil.checkData(value);
            }
        });
        //8.数据转换、合并
        DataStream <CleanBean> sseCleanBeanMap = sseFileData.map(new SseMap());
        DataStream <CleanBean> szseCleanBeanMap = szseFileData.map(new SzseMap());
        DataStream<CleanBean> cleanBeanDataStream = sseCleanBeanMap.union(szseCleanBeanMap);
        //9.过滤指数数据
        DataStream<CleanBean> cleanBeanFilter = cleanBeanDataStream.filter(new FilterFunction<CleanBean>() {
            @Override
            public boolean filter(CleanBean value) throws Exception {
                return QuotUtil.isIndex(value);
            }
        });
        DataStream<CleanBean> cleanBean = cleanBeanFilter.assignTimestampsAndWatermarks(new BoundedOutOfOrdernessTimestampExtractor<CleanBean>(Time.seconds(Long.parseLong(QuotConfig.config.getProperty("delay.time")))) {
            @Override
            public long extractTimestamp(CleanBean element) {
                return element.getEventTime();
            }
        });

        //秒级行情
        new IndexQuotSecTask().process(cleanBean);
        //分时行情
        new IndexQuotMinTask().process(cleanBean);
        //3.分时数据备份
        new IndexQuotMinHdfsTask().process(cleanBean);

        try {
            env.execute("index stream");
        } catch (Exception e) {
            e.printStackTrace();
        }
    }
}
